package cloud.coder.remote.grpc.client.service;

import cloud.coder.remote.boot.BootService;
import cloud.coder.remote.boot.DefaultImplementor;
import cloud.coder.remote.boot.ServiceManager;
import cloud.coder.remote.config.Config;
import cloud.coder.remote.grpc.client.GRPCChannelListener;
import cloud.coder.remote.grpc.client.GRPCChannelManager;
import cloud.coder.remote.grpc.client.GRPCChannelStatus;
import cloud.coder.remote.grpc.client.GRPCStreamServiceStatus;
import io.grpc.Channel;
import io.grpc.StatusRuntimeException;
import io.grpc.examples.msg.report.MsgCollection;
import io.grpc.examples.msg.report.MsgObject;
import io.grpc.examples.msg.report.MsgReportServiceGrpc;
import io.grpc.examples.network.common.Commands;
import io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.TimeUnit;

@DefaultImplementor
public class MsgReportServiceClient implements BootService, GRPCChannelListener {
    private static final Logger LOGGER = LoggerFactory.getLogger(MsgReportServiceClient.class);

    private volatile GRPCChannelStatus status = GRPCChannelStatus.DISCONNECT;
    private volatile MsgReportServiceGrpc.MsgReportServiceStub serviceStub;
    private volatile MsgReportServiceGrpc.MsgReportServiceBlockingStub serviceBlockingStub;

    private long lastSendTime;
    private long msgUplinkedCounter;
    private long msgAbandonedCounter;

    @Override
    public void prepare() {
        ServiceManager.INSTANCE.findService(GRPCChannelManager.class).addChannelListener(this);
    }

    @Override
    public void boot() {
        lastSendTime = System.currentTimeMillis();
        msgUplinkedCounter = 0;
        msgAbandonedCounter = 0;
    }

    @Override
    public void onComplete() {

    }

    @Override
    public void shutdown() {

    }

    public void reportInSync(MsgCollection msgCollection) throws InterruptedException {
        Commands commands;
        if (status != GRPCChannelStatus.CONNECTED) {
            TimeUnit.SECONDS.sleep(1);
        }
        try {
            commands = serviceBlockingStub.reportInSync(msgCollection);
        } catch (StatusRuntimeException e) {
            LOGGER.warn("RPC failed: {0}", e.getStatus());
            LOGGER.error(e.getMessage(), e);
            return;
        }
        LOGGER.info(commands.toString());
    }

    public void reportBySteam(List<MsgObject> data) throws InterruptedException {
        if (GRPCChannelStatus.CONNECTED.equals(status)) {
            final GRPCStreamServiceStatus status = new GRPCStreamServiceStatus(false);
            StreamObserver<MsgObject> upstreamSegmentStreamObserver = serviceStub.withDeadlineAfter(
                    Config.GRPC_UPSTREAM_TIMEOUT, TimeUnit.SECONDS
            ).report(new StreamObserver<Commands>() {
                @Override
                public void onNext(Commands commands) {
                    //TODO
                    LOGGER.info(commands.toString());
                }

                @Override
                public void onError(
                        Throwable throwable) {
                    status.finished();

                    LOGGER.error(
                            "Send UpstreamSegment to collector fail with a grpc internal exception.", throwable
                    );
                    ServiceManager.INSTANCE
                            .findService(GRPCChannelManager.class)
                            .reportError(throwable);
                }

                @Override
                public void onCompleted() {
                    status.finished();
                }
            });

            for (MsgObject msg : data) {
                upstreamSegmentStreamObserver.onNext(msg);
            }


            upstreamSegmentStreamObserver.onCompleted();

            status.wait4Finish();
            msgUplinkedCounter += data.size();
        } else {
            msgAbandonedCounter += data.size();
        }

        printUplinkStatus();
    }

    private void printUplinkStatus() {
        long currentTimeMillis = System.currentTimeMillis();
        if (currentTimeMillis - lastSendTime > 30 * 1000) {
            lastSendTime = currentTimeMillis;
            if (msgUplinkedCounter > 0) {
                LOGGER.debug("{} msg have been sent to server.", msgUplinkedCounter);
                msgUplinkedCounter = 0;
            }
            if (msgAbandonedCounter > 0) {
                LOGGER.debug(
                        "{} msg have been abandoned, cause by no available channel.", msgAbandonedCounter);
                msgAbandonedCounter = 0;
            }
        }
    }


    @Override
    public void statusChanged(GRPCChannelStatus status) {
        if (GRPCChannelStatus.CONNECTED.equals(status)) {
            Channel channel = ServiceManager.INSTANCE.findService(GRPCChannelManager.class).getChannel();
            serviceStub = MsgReportServiceGrpc.newStub(channel);
            serviceBlockingStub = MsgReportServiceGrpc.newBlockingStub(channel);
        }
        this.status = status;
    }
}
